/*
 * Copyright (C) 2010 TopCoder Inc., All Rights Reserved.
 */
package com.hercules.xre.protocol.testing.netty.tcp;
import java.nio.charset.UnsupportedCharsetException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import com.topcoder.util.log.Log;
import com.topcoder.util.log.LogManager;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipelineCoverage;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;

import com.hercules.xre.protocol.testing.Helper;
import com.hercules.xre.protocol.testing.netty.CommandSender;
import com.hercules.xre.protocol.testing.netty.CommandsDeliveryStatusFutureListener;
import com.hercules.xre.protocol.testing.netty.ExceptionHandlingChannelFutureListener;
import com.hercules.xre.protocol.testing.netty.IllegalMessageException;
import com.hercules.xre.protocol.testing.netty.NoSuchSessionException;
import com.hercules.xre.protocol.testing.EventHandler;
import com.hercules.xre.protocol.testing.EventHandlerException;
/**
 * <p>
 * Channel handler for messages passed trough TCP connection.
 * This handler should be proceeded in the pipeline by frame decoder handler (TCPFrameDecoder).
 * </p>
 * <p>
 * For incoming messages this class is responsible for decoding of string representing passed
 * XRE events. Decoded string is passed to the next handler in the pipeline.
 * For messages that need to be sent this class is responsible for serialization of passed
 * commands and encoding them according to specification in 2.6 of XRE Protocol Specification
 * (first 4 bytes of passed message should contain passed message length).
 * Also it is responsible for monitoring if passed commands have been delivered successfully.
 * </p>
 *
 * <p>
 * <strong>Thread-safety:</strong> This class is thread safe.
 * Thread safety is achieved by additional synchronization of mutable fields access
 * commandsBeingSent is a synchronized set
 * Also connectionManager is used but only following methods are used:
 * getSessionGUIDByChannel()
 * removeChannelSessionGUIDAssociation() and all of mentioned methods use internal synchronization.
 * </p>
 * @author TCSDESIGNER, TCSDEVELOPER
 * @version 1.0
 */
@ChannelPipelineCoverage("one")
public class TCPMessageHandler extends SimpleChannelHandler implements CommandSender {
    /**
     * <p>
     * Logger that is used to log activities and errors.
     * It is initialized to LogManager.getLog(TCPMessageHandler.class.getName()).
     * Can not be null, never changes after initialization.
     * </p>
     */
    private static final Log LOGGER = LogManager.getLog(TCPMessageHandler.class.getName());

    /**
     * <p>
     * Synchronized set that holds all commands that have been written to the channel but
     * no Netty API hasn't reported yet that delivery was successful.
     * It is used to report application about commands that haven't been sent in case of connection failure.
     * </p>
     * <p>
     * Field is modified in +writeRequest() and +commandsDelivered() methods.
     * Field never changes after initialization (set contents may change).
     * </p>
     */
    private final Set<String> commandsBeingSent = Collections.synchronizedSet(new HashSet<String>());

    /**
     * <p>
     * Connection eventHandler that is used to inform application about communication events.
     * Set in the constructor, never changes after that. Can't contain null.
     * </p>
     */
    private final EventHandler eventHandler;

    /**
     * <p>
     * Connection manager that is used to fetch sessionGUID for current channel or
     * to remove association between channel and a sessionGUID (when channel has been disconnected).
     * Set in the constructor, never changes after that. Can't contain null.
     * </p>
     */
    private final TCPProtocolTestingConnectionManager  connectionManager;

    /**
     * <p>
     * Creates new instance.
     * </p>
     * @param eventHandler event handler that will be used to report detected events, can't be null
     * @param connectionManager connection manager, can't be null
     * @throws IllegalArgumentException when any attribute is invalid
     */
    public TCPMessageHandler(TCPProtocolTestingConnectionManager connectionManager, EventHandler eventHandler) {
        final String signature = "TCPMessageHandler#ctor";
        Helper.logEnter(signature, new String[] {"connectionManager", "eventHandler"},
            new Object[] {connectionManager, eventHandler}, LOGGER);
        try {
            Helper.checkNull(connectionManager, "connectionManager", signature, LOGGER);
            Helper.checkNull(eventHandler, "eventHandler", signature, LOGGER);
            this.connectionManager = connectionManager;
            this.eventHandler = eventHandler;
        } finally {
            Helper.logExit(signature, null, LOGGER);
        }
    }

    /**
     * <p>
     * Method called when new message has been received and decoded.
     * It creates new String from bytes passed in message, this newly
     * created string is then passed to the next handler in pipeline.
     * </p>
     * @param e message event, can't be null
     * @param ctx channel handler context, can't be null
     * @throws IllegalArgumentException when any of parameters is invalid.
     * @throws IllegalMessageException when passed message is not a ChannelBuffer
     */
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws IllegalMessageException {
        final String signature = "TCPMessageHandler#messageReceived";
        Helper.logEnter(signature, new String[] {"ChannelHandlerContext", "MessageEvent"},
            new Object[] {ctx, e}, LOGGER);
        try {
            Helper.checkNull(ctx, "ctx", signature, LOGGER);
            Helper.checkNull(e, "MessageEvent", signature, LOGGER);
            // get string representation of ChannelBuffer content contained within message event
            Object message = e.getMessage();
            if (!(message instanceof ChannelBuffer)) {
                throw Helper.logError(new IllegalMessageException(
                    "the passed message is not a ChannelBuffer"), signature, LOGGER);
            }
            if (ctx.getChannel() == null) {
                // Note : Channels#write will throw NPE if future is null. So we must check it first
                throw Helper.logError(new IllegalMessageException(
                    "Channel of ChannelHandlerContext can't be null"), signature, LOGGER);
            }
            try {
                // fire message received event on next handler in pipeline with retrieved message string as an event
                // message
                Channels.fireMessageReceived(ctx, ((ChannelBuffer) message).toString("UTF-8"));
            } catch (UnsupportedCharsetException ex) {
                // should never occurs
                throw Helper.logError(new IllegalMessageException(
                    "unsupported utf-8 charset", ex), signature, LOGGER);
            }
        } finally {
            Helper.logExit(signature, null, LOGGER);
        }
    }

    /**
     * <p>
     * Method called when new command is to be sent to a client.
     * Passed command is serialized and new message is composed according to section 2.6
     * in XRE Protocol Specification. After that message is passed for sending.
     * </p>
     * <p>
     * In order to trace commands send status "commandsBeingSent" set is used.
     * When new command is sent it is added to this set and new FutureListener
     * is created that monitors if has been properly sent. If it has, then command
     * will be removed from set in #commandsDelivered method. If it hasn't then probably
     * there was a connection error and application will be informed about delivery
     * failure by +channelDisconnected() method.
     * </p>
     * @param e message event, can't be null
     * @param ctx channel handler context, can't be null
     * @throws IllegalArgumentException when any of parameters is invalid.
     * @throws IllegalMessageException if the message is invalid.
     * @throws NoSuchSessionException when no session was found.
     */
    public void writeRequested(ChannelHandlerContext ctx, MessageEvent e)
        throws IllegalMessageException, NoSuchSessionException {
        final String signature = "TCPMessageHandler#writeRequested";
        Helper.logEnter(signature, new String[] {"ChannelHandlerContext", "MessageEvent"},
            new Object[] {ctx, e}, LOGGER);
        try {
            Helper.checkNull(ctx, "ctx", signature, LOGGER);
            Helper.checkNull(e, "MessageEvent", signature, LOGGER);
            String command = (String) e.getMessage();
            Channel channel = ctx.getChannel();
            if (channel == null) {
                throw Helper.logError(new IllegalMessageException(
                    "Channel of ChannelHandlerContext can't be null"), signature, LOGGER);
            }
            // replace command index and timestamp
            try {
                command = connectionManager.prepareCommandToSend(command, channel);
            } catch (IllegalMessageException ex) {
                throw Helper.logError(ex, signature, LOGGER);
            }
            byte[] commandBytes = command.getBytes();
            // create new buffer for entire message (4 bytes for header and entire command string)
            ChannelBuffer buffer = ChannelBuffers.buffer(4 + commandBytes.length);

            // write length of message that is being sent
            buffer.writeInt(commandBytes.length);

            // write rest of the message
            buffer.writeBytes(commandBytes);

            // pass buffer to next handler in the pipeline
            ChannelFuture future = e.getFuture();

            if (future == null) {
                throw Helper.logError(new IllegalMessageException(
                    "future can not be null"), signature, LOGGER);
            }

            Channels.write(ctx, future, buffer);
            // add listener to handle any exceptions
            future.addListener(new ExceptionHandlingChannelFutureListener());

            // Add commands to set of commands that are being
            // sent to control if it has been successfully delivered
            commandsBeingSent.add(command);

            // add future listener that will monitor command delivery status
            future.addListener(new CommandsDeliveryStatusFutureListener(
                this, Arrays.asList(command)));
        } catch (NoSuchSessionException ex) {
            throw Helper.logError(ex, signature, LOGGER);
        } finally {
            Helper.logExit(signature, null, LOGGER);
        }
    }

    /**
     * <p>
     * Channel state event handler called when channel has been disconnected from its peer.
     * This method informs (using eventHandler) application about encountered situation.
     * </p>
     * @param e message event, can't be null
     * @param ctx channel handler context, can't be null
     * @throws IllegalArgumentException when any of parameters is invalid.
     * @throws NoSuchSessionException if no specific session was found.
     */
    public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws NoSuchSessionException {
        final String signature = "TCPMessageHandler#channelDisconnected";
        Helper.logEnter(signature, new String[] {"ChannelHandlerContext", "ChannelStateEvent"},
            new Object[] {ctx, e}, LOGGER);
        try {
            Helper.checkNull(ctx, "ctx", signature, LOGGER);
            Helper.checkNull(e, "ChannelStateEvent", signature, LOGGER);
            Channel channel = ctx.getChannel();
            if (channel == null) {
                throw Helper.logError(new NoSuchSessionException(
                    "can't found session guid for the channel"), signature, LOGGER);
            }

            String sessionGUID;
            try {
                sessionGUID = connectionManager.getSessionGUIDByChannel(channel);
            } catch (NoSuchSessionException ex) {
                throw Helper.logError(ex, signature, LOGGER);
            }

            List<String> commandsNotDelivered = new ArrayList<String>();
            synchronized (commandsBeingSent) {
                commandsNotDelivered.addAll(commandsBeingSent);
            }

            // call event handler
            eventHandler.handleConnectionDisconnected(sessionGUID, commandsNotDelivered);
            // remove channel from list of active channels
            connectionManager.removeChannelSessionGUIDAssociation(channel);
        } catch (NoSuchSessionException ex) {
            throw Helper.logError(ex, signature, LOGGER);
        } catch (EventHandlerException ex) {
            Helper.logError(ex, signature, LOGGER);
        } finally {
            Helper.logExit(signature, null, LOGGER);
        }
    }

    /**
     * <p>
     * Method called when command have been successfully delivered.
     * </p>
     * @param commands list of commands that have been delivered, can't be empty
     *        can't contain null/empty command.
     * @throws IllegalArgumentException - when passed parameter is null,empty,
     *         or contains null/empty command.
     */
    public void commandsDelivered(List<String> commands) {
        final String signature = "TCPMessageHandler#commandsDelivered";
        Helper.logEnter(signature, new String[] {"commands"}, new Object[] {commands}, LOGGER);
        try {
            Helper.checkList(commands, "commands", signature, LOGGER);
            commandsBeingSent.removeAll(commands);
        } finally {
            Helper.logExit(signature, null, LOGGER);
        }
    }
}

